package com.boyi.center.service.impl;


import com.boyi.center.service.UserService;
import com.boyi.center.task.KafkaTask;
import com.boyi.db.entity.User;
import com.boyi.db.mapper.UserMapper;
import com.boyi.shared.common.Result;
import com.boyi.shared.kafka.KafkaResult;
import lombok.extern.log4j.Log4j2;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.apache.kafka.clients.producer.ProducerRecord;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.stereotype.Service;
import org.springframework.web.client.RestTemplate;

import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.concurrent.TimeUnit;

/**
 * @author tingwen
 * @date 2017年12月28日  23:59
 */
@Service
@Log4j2
public class UserServiceImpl extends BaseServiceImpl<User> implements UserService {
    @Autowired
    private UserMapper userMapper;
    @Autowired
    private KafkaTask kafkaTask;

    @Autowired
    private RestTemplate restTemplate;

    @Override
    public Result<List<User>> save(User entity) {
        //存库
        Result<List<User>> save = super.save(entity);
        //存redis
        //发送kafka
//        KafkaResult<Object> kafkaResult = new KafkaResult<>(entity);
//        ProducerRecord<String, KafkaResult<Object>> record = new ProducerRecord<>("fooTest", kafkaResult);
//        kafkaTask.taskPublic(record);
        return save;
    }

    @Override
    public void updateStateByPrimaryKey(String userId, Short state) {
        User user = new User();
        user.setUserId(userId);
        user.setState(state);
        Short oldState = userMapper.selectStateByPrimaryKey(userId);
        if (!state.equals(oldState)) {
            mapper.updateByPrimaryKeySelective(user);
        }
    }

    /*---------------- 注册功能消费者--start ---------------------*/
    private final static Map<Integer, Long> WAIT_TIME_HASH_MAP = new HashMap<>();

    static {
        WAIT_TIME_HASH_MAP.put(0, 0L);
        WAIT_TIME_HASH_MAP.put(1, (long) 0.5);
        WAIT_TIME_HASH_MAP.put(2, 1L);
        WAIT_TIME_HASH_MAP.put(3, 2L);
        WAIT_TIME_HASH_MAP.put(4, 3L);
    }

//    @KafkaListener(topics = "footest", concurrency = "10")
    private void listen(ConsumerRecord<?, KafkaResult<Object>> record) {
        KafkaResult<Object> kafkaResult = record.value();
        User user = (User) kafkaResult.getData();
        int notifyTime = kafkaResult.getNotifyTime();
        consumerHandle(user, notifyTime);
    }

//    @KafkaListener(topics = "test", concurrency = "1")
    private void listen1(ConsumerRecord<?, String> record) {
        log.error(record.value());
    }

    private void consumerHandle(User user, int notifyTime) {
        int retryTime = 5;
        while (notifyTime < retryTime) {
            Long factor = WAIT_TIME_HASH_MAP.get(notifyTime);
            String str;
            try {
                TimeUnit.MINUTES.sleep(factor);
                str = restTemplate.postForObject(user.getNotifyUrl(), null, String.class);
                if ("success".equals(str)) {
                    updateStateByPrimaryKey(user.getUserId(), (short) 1);
                }
                notifyTime = 5;
            } catch (Exception e) {
                logger.error("RestClientException", e);
                notifyTime++;
            }
        }
    }
    /*---------------- 注册功能消费者--end ---------------------*/

}
